﻿# coding=utf-8
# 2023/4/27: 首个记录，基于旧2023/3/21版本
# 2023/4/28: 修正datetime模块导入问题
# 2023/5/4: 添加注释，支持自动输出文档
# 2023/5/6: 增加StandaloneAgency
# 2023/6/19: API改为standalone_before_start，standalone_before/after_loop，standalone_after_end。提供完备的IO功能
# 2023/8/23: 修正获取script_id问题
# 2024/2/28: 修正信号表读取问题
# 2025/7/18: 使用类型注解，修正若干逻辑

""" 此模块可支持以独立模式运行

独立运行模式用于在VSCode、PyCharm或其他IDE工具下开发脚本，调试通过后再导入应用程序运行，可提高开发效率。
注意，运行前需要将bi_common.py, bi_standalone.py, 以及相关样本代码文件拷贝至工作空间。
"""

import os
import datetime
import math
import glob
import xml.dom.minidom
from typing import Dict, List, Any, TextIO
from enum import Enum
import bi_common as bi


class _SignalInConfig:
    def __init__(self, name: str, source: str, nearest: bool, default_val: float):
        self.name: str = name
        self.source: str = source
        self.nearest: bool = nearest
        self.default_val: float = default_val


class _SampleOutConfig:
    def __init__(self, protocol: str, title: str):
        self.protocol: str = protocol
        self.title: str = title


class _GraphType(Enum):
    INVALID = 0
    SINGLE_VALUE = 1
    SCATTER_POINTS = 2
    HIST_AND_LINE = 3
    MATRIX_TABLE = 4
    LABEL_TABLE = 5

    @staticmethod
    def to_camel_string(type: '_GraphType') -> str:
        if type == _GraphType.SINGLE_VALUE:
            return "SingleValue"
        elif type == _GraphType.SCATTER_POINTS:
            return "ScatterPoints"
        elif type == _GraphType.HIST_AND_LINE:
            return "HistAndLine"
        elif type == _GraphType.MATRIX_TABLE:
            return "MatrixTable"
        elif type == _GraphType.LABEL_TABLE:
            return "LabelTable"
        else:
            return "Invalid"
        
    @staticmethod
    def from_camel_string(str: str) -> '_GraphType':
        if str == "SingleValue":
            return _GraphType.SINGLE_VALUE
        elif str == "ScatterPoints":
            return _GraphType.SCATTER_POINTS
        elif str == "HistAndLine":
            return _GraphType.HIST_AND_LINE
        elif str == "MatrixTable":
            return _GraphType.MATRIX_TABLE
        elif str == "LabelTable":
            return _GraphType.LABEL_TABLE
        else:
            return _GraphType.INVALID


class _ReportOutConfig:
    def __init__(self, id: str, title: str, type: _GraphType, column_titles: List[str], configs: List[str]):
        self.id: str = id
        self.title: str = title
        self.type: _GraphType = type
        self.column_titles: List[str] = column_titles
        self.configs: List[str] = configs


class _ReportData:
    def __init__(self, config: '_ReportOutConfig'):
        self.config: _ReportOutConfig = config
        self.params: List[str] = []
        self.data: List[List[int | float]] = []


class _SingleValueReportData(_ReportData):
    def __init__(self, config: _ReportOutConfig):
        super().__init__(config)
        self.data = [[0.0]]

    def update_report(self, parameters: Dict[str, Any]):
        if "x_set" in parameters:
            self.data[0][0] = parameters["x_set"]
        elif "x" in parameters:
            self.data[0][0] = self.data[0][0] + parameters["x"]


class _ScatterPointsReportData(_ReportData):
    def __init__(self, config: _ReportOutConfig):
        super().__init__(config)
        self.params = ["0"]
        self.data = []

    def update_report(self, parameters: Dict[str, Any]):
        if "x" not in parameters or "y" not in parameters:
            return
        x = parameters["x"]
        y = parameters["y"]
        x_lower = float(self.config.configs[0])
        x_upper = float(self.config.configs[1])
        y_lower = float(self.config.configs[2])
        y_upper = float(self.config.configs[3])
        if x < x_lower or x > x_upper or y < y_lower or y > y_upper:
            return
        if "t" in parameters and "session" in parameters:
            sess = datetime.datetime.strptime(parameters["session"], "%Y%m%d%H%M%S")
            posix = float((sess - datetime.datetime(1970, 1, 1, 0, 0, 0)).seconds)
            t = parameters["t"]
            self.data.append([x, y, posix, t])
        else:
            self.data.append([x, y, 0, 0])
        self.params[0] = str(len(self.data))


class _HistAndLineReportData(_ReportData):
    def __init__(self, config: _ReportOutConfig):
        super().__init__(config)
        rows = 0
        if config.configs[4] == "XValues":
            rows = int(config.configs[7])
        elif config.configs[4] == "XLabels":
            rows = len(config.configs) - 5
        line_enabled = config.configs[0] not in ["Sum", "Aver", "Min", "Max", "Percentage", "HitRatio"]
        default_hist_value = int(config.configs[2])
        default_line_value = int(config.configs[3])
        self.data = []
        for i in range(0, rows):
            if line_enabled:
                self.data.append([default_hist_value, 0, default_line_value, 0])
            else:
                self.data.append([default_hist_value, 0])

    def update_report(self, parameters: Dict[str, Any]):
        if "hit" in parameters and "times" in parameters:
            hit = 1 if parameters["hit"] else 0
            times = parameters["times"]
            if "x" in parameters:
                self._add(parameters["x"], hit, None, times, False, True)
            elif "x_index" in parameters:
                self._add(parameters["x_index"], hit, None, times, True, True)
        elif "y" in parameters and "y2" in parameters:
            y = parameters["y"]
            y2 = parameters["y2"]
            if "x" in parameters:
                self._add(parameters["x"], y, y2, 1, False, False)
            elif "x_index" in parameters:
                self._add(parameters["x_index"], y, y2, 1, True, False)

    def _add(self, x: float | int, y: float, y2: float | None, times: int, x_is_index: bool, add_hit: bool):
        if not x_is_index and self.config.configs[4] != "XValues":
            return
        if times <= 0:
            return
        line_enabled = self.config.configs[0] not in ["Sum", "Aver", "Min", "Max", "Percentage", "HitRatio"]
        if y2 is not None and not line_enabled:
            return
        if add_hit and not line_enabled:
            return
        if add_hit and y != 0:
            y = 100
        if y2 is None:
            y2 = y
        x_index: int = 0
        if x_is_index:
            x_index = int(x)
            if x_index < 0 or x_index >= len(self.data):
                return
        else:
            bas = float(self.config.configs[5])
            step = float(self.config.configs[6])
            count = float(self.config.configs[7])
            x_index = int((x - bas) / step)
            if x_index < 0 or x_index >= count:
                return
        mode = self.config.configs[0]
        if mode == "AverAndDev":
            for i in range(0, times):
                aver = self.data[x_index][0]
                count = self.data[x_index][1]
                dev = self.data[x_index][2]
                if count == 0:
                    self.data[x_index][0] = y
                    self.data[x_index][1] = 1
                    self.data[x_index][2] = 0
                else:
                    count_before = count
                    count_after = count + 1
                    sum_before = aver * count_before
                    sum_after = sum_before + y
                    sumv = dev * dev * count_before + sum_before * sum_before / count_before + y * y
                    vara = (sumv - sum_after * sum_after / count_after) / count_after
                    dev_final = math.sqrt(vara) if vara > 0 else 0
                    self.data[x_index][0] = sum_after / count_after
                    self.data[x_index][1] = count_after
                    self.data[x_index][2] = dev_final
        else:
            if mode == "Sum" or mode == "Percentage" or mode == "SumAndSum":
                self.data[x_index][0] = self.data[x_index][0] + y * times
            elif mode == "Aver" or mode == "HitRatio":
                self.data[x_index][0] = (self.data[x_index][0] * self.data[x_index][1] + y * times) / (self.data[x_index][1] + times)
                self.data[x_index][1] = self.data[x_index][1] + times
            elif mode == "Min" or mode == "MinAndMin" or mode == "MinAndMax":
                if self.data[x_index][1] == 0:
                    self.data[x_index][0] = y
                    self.data[x_index][1] = 1
                else:
                    self.data[x_index][0] = min(self.data[x_index][0], y)
            elif mode == "Max" or mode == "MaxAndMax":
                if self.data[x_index][1] == 0:
                    self.data[x_index][0] = y
                    self.data[x_index][1] = 1
                else:
                    self.data[x_index][0] = max(self.data[x_index][0], y)
            elif mode == "AverAndMin" or mode == "AverAndMax" or mode == "AverAndAver":
                self.data[x_index][0] = (self.data[x_index][0] * self.data[x_index][1] + y * times) / (self.data[x_index][1] + times)
                self.data[x_index][1] = self.data[x_index][1] + times
            if mode == "SumAndSum":
                self.data[x_index][2] = self.data[x_index][2] + y2 * times
            elif mode == "MinAndMin" or mode == "AverAndMin":
                if self.data[x_index][3] == 0:
                    self.data[x_index][2] = y2
                    self.data[x_index][3] = 1
                else:
                    self.data[x_index][2] = min(self.data[x_index][2], y2)
            elif mode == "MinAndMax" or mode == "MaxAndMax" or mode == "AverAndMax":
                if self.data[x_index][3] == 0:
                    self.data[x_index][2] = y2
                    self.data[x_index][3] = 1
                else:
                    self.data[x_index][2] = max(self.data[x_index][2], y2)
            elif mode == "AverAndAver":
                self.data[x_index][2] = (self.data[x_index][2] * self.data[x_index][3] + y2 * times) / (self.data[x_index][3] + times)
                self.data[x_index][3] = self.data[x_index][3] + times


class _MatrixTableReportData(_ReportData):
    def __init__(self, config: _ReportOutConfig):
        super().__init__(config)
        x_count = int(config.configs[3])
        y_count = int(config.configs[6])
        default_val = float(config.configs[11])
        self.data = []
        for x in range(0, x_count):
            row = []
            for y in range(0, y_count):
                row.append(default_val)
                row.append(0)
            self.data.append(row)

    def update_report(self, parameters: Dict[str, Any]):
        if "x" not in parameters or "y" not in parameters or "z" not in parameters:
            return
        x = parameters["x"]
        y = parameters["y"]
        z = parameters["z"]
        x_base = float(self.config.configs[1])
        x_step = float(self.config.configs[2])
        x_count = int(self.config.configs[3])
        y_base = float(self.config.configs[4])
        y_step = float(self.config.configs[5])
        y_count = int(self.config.configs[6])
        x = int((x - x_base) / x_step)
        y = int((y - y_base) / y_step)
        if x < 0 or x >= x_count or y < 0 or y >= y_count:
            return
        mode = self.config.configs[0]
        if mode == "Sum" or mode == "Percentage":
            self.data[x][y * 2] = self.data[x][y * 2] + z
        elif mode == "Min":
            if self.data[x][y * 2 + 1] == 0:
                self.data[x][y * 2] = z
                self.data[x][y * 2 + 1] = 1
            else:
                self.data[x][y * 2] = min(self.data[x][y * 2], z)
        elif mode == "Max":
            if self.data[x][y * 2 + 1] == 0:
                self.data[x][y * 2] = z
                self.data[x][y * 2 + 1] = 1
            else:
                self.data[x][y * 2] = max(self.data[x][y * 2], z)
        elif mode == "Aver":
            self.data[x][y * 2] = (self.data[x][y * 2] * self.data[x][y * 2 + 1] + z) / (self.data[x][y * 2 + 1] + 1)
            self.data[x][y * 2 + 1] = self.data[x][y * 2 + 1] + 1


class _LabelTableReportData(_ReportData):
    def __init__(self, config: _ReportOutConfig):
        super().__init__(config)
        x_count = int(config.configs[1])
        y_count = int(config.configs[2])
        default_val = float(config.configs[4])
        self.data = []
        for x in range(0, x_count):
            row = []
            for y in range(0, y_count):
                row.append(default_val)
                row.append(0)
            self.data.append(row)

    def update_report(self, parameters: Dict[str, Any]):
        if "x_index" not in parameters or "y_index" not in parameters or "z" not in parameters:
            return
        x = parameters["x_index"]
        y = parameters["y_index"]
        z = parameters["z"]
        x_count = int(self.config.configs[1])
        y_count = int(self.config.configs[2])
        if x < 0 or x >= x_count or y < 0 or y >= y_count:
            return
        mode = self.config.configs[0]
        if mode == "Sum" or mode == "Percentage":
            self.data[x][y * 2] = self.data[x][y * 2] + z
        elif mode == "Min":
            if self.data[x][y * 2 + 1] == 0:
                self.data[x][y * 2] = z
                self.data[x][y * 2 + 1] = 1
            else:
                self.data[x][y * 2] = min(self.data[x][y * 2], z)
        elif mode == "Max":
            if self.data[x][y * 2 + 1] == 0:
                self.data[x][y * 2] = z
                self.data[x][y * 2 + 1] = 1
            else:
                self.data[x][y * 2] = max(self.data[x][y * 2], z)
        elif mode == "Aver":
            self.data[x][y * 2] = (self.data[x][y * 2] * self.data[x][y * 2 + 1] + z) / (self.data[x][y * 2 + 1] + 1)
            self.data[x][y * 2 + 1] = self.data[x][y * 2 + 1] + 1


class _SceneOutConfig:
    def __init__(self, id: str, title: str):
        self.id = id
        self.title = title


class _ConfigIO:
    def __init__(self, path: str, script_id: str, globals: Dict[str, Any]):
        blist = [ "bi", "bis", "session_dirs", "generations", "session_index", "generation_index", "global_params", "global_vars", "frequency" ]
        replacements = {}
        for key in globals:
            if key.startswith("__"):
                continue
            if key in blist:
                continue
            replacements["{" + key + "}"] = str(globals[key])

        self.signal_in: List[_SignalInConfig] = []
        self.signal_out: List[str] = []
        self.sample_in: List[str] = []
        self.sample_out: List[_SampleOutConfig] = []
        self.report_out: List[_ReportOutConfig] = []
        self.scene_out: List[_SceneOutConfig] = []

        dom = xml.dom.minidom.parse(path)
        root_node = dom.documentElement
        if root_node is None:
            return

        for sub_node in root_node.getElementsByTagName("in"):
            name = sub_node.getAttribute("param")
            source = self._replace(sub_node.getAttribute("signal"), replacements)
            nearest = sub_node.getAttribute("nearest") == "yes"
            default_val = float(sub_node.getAttribute("default_val"))
            self.signal_in.append(_SignalInConfig(name, source, nearest, default_val))

        for sub_node in root_node.getElementsByTagName("out"):
            self.signal_out.append(sub_node.getAttribute("param"))

        for sub_node in root_node.getElementsByTagName("sample_in"):
            if sub_node.firstChild is not None:
                self.sample_in.append(self._replace(sub_node.firstChild.data, replacements))  # type: ignore

        for sub_node in root_node.getElementsByTagName("sample_out"):
            protocol = self._replace(sub_node.getAttribute("id"), replacements)
            title = sub_node.firstChild.data if sub_node.firstChild is not None else ""  # type: ignore
            self.sample_out.append(_SampleOutConfig(protocol, title))

        for sub_node in root_node.getElementsByTagName("report_out"):
            id = sub_node.getAttribute("id")
            title = sub_node.getAttribute("title") + " (" + script_id + ")"
            type = _GraphType.from_camel_string(sub_node.getAttribute("type"))
            column_texts = sub_node.getAttribute("column_titles")
            column_titles = [] if len(column_texts) == 0 else column_texts.split(",")
            config_texts = sub_node.getAttribute("configs")
            configs = [] if len(config_texts) == 0 else config_texts.split(",")
            self.report_out.append(_ReportOutConfig(id, title, type, column_titles, configs))

        for sub_node in root_node.getElementsByTagName("scene_out"):
            id = self._replace(sub_node.getAttribute("id"), replacements)
            title = sub_node.firstChild.data if sub_node.firstChild is not None else ""  # type: ignore
            self.scene_out.append(_SceneOutConfig(id, title))
    
    def _replace(self, input: str, replacements: Dict[str, str]) -> str:
        for key in replacements:
            input = input.replace(key, replacements[key])
        return input


class _SessionMeta:
    def __init__(self, path: str):
        self.session_id: datetime.datetime = datetime.datetime.min
        self.length: float = 0.0
        dom = xml.dom.minidom.parse(path)
        root_node = dom.documentElement
        if root_node is None:
            return
        comps = root_node.getAttribute("session_id").split("-")
        self.session_id = datetime.datetime(int(comps[0]), int(comps[1]), int(comps[2]), int(comps[3]), int(comps[4]), int(comps[5]))
        self.length = float(root_node.getAttribute("length"))


class _GenerationInfo:
    def __init__(self, path: str):
        self.gen_id: str = ""
        dom = xml.dom.minidom.parse(path)
        root_node = dom.documentElement
        if root_node is None:
            return
        self.gen_id = root_node.getAttribute("gen_id")


class _SignalElem:
    def __init__(self, time: float, val: float):
        self.time: float = time
        self.val: float = val


class _SignalLoader:
    def __init__(self, signal_csv_path: str, category: str):
        self._valid: bool = False
        self._file: TextIO | None = None
        self._category: str = ""
        self._type: str = ""
        self._buffers: Dict[str, List[_SignalElem]] = {}
        self._last_time: float = 0.0
        file = open(signal_csv_path)
        type = self._read_type(file)
        if type is not None:
            self._valid = True
            self._file = file
            self._category = category
            self._type = type
        else:
            file.close()

    def load(self, target: float, max_gap: float):
        if not self._valid or self._file is None:
            return None
        if self._last_time < target + max_gap:
            self._last_time = max(self._last_time, self._read_to_buffers(self._file, target + max_gap, self._category, self._type, self._buffers))
        self._clip_buffers(target - max_gap, self._buffers)

    def signal_ids(self) -> List[str]:
        if not self._valid:
            return []
        else:
            return list(self._buffers.keys())

    def get_val(self, target: float, id: str, nearest: bool, default_val: float) -> float:
        if not self._valid or id not in self._buffers:
            return default_val
        buffer = self._buffers[id]
        if nearest:
            if len(buffer) == 0:
                return default_val
            diff = None
            val = default_val
            for elem in buffer:
                cur_diff = abs(elem.time - target)
                if diff is None or cur_diff < diff:
                    diff = cur_diff
                    val = elem.val
            return val
        else:
            if len(buffer) < 2:
                return default_val
            for i in range(0, len(buffer) - 1):
                delta = buffer[i + 1].time - buffer[i].time
                if delta <= 0:
                    continue
                if buffer[i].time <= target and buffer[i + 1].time > target:
                    w1 = (buffer[i + 1].time - target) / delta
                    w2 = (target - buffer[i].time) / delta
                    return buffer[i].val * w1 + buffer[i + 1].val * w2
            return default_val

    def close(self):
        if self._file is not None:
            self._file.close()
            self._file = None
        self._valid = False

    def _read_type(self, file: TextIO) -> str | None:
        elems = file.readline().rstrip("\n").split(",")
        if len(elems) != 3 or not elems[0].endswith("ignal Table") or elems[1] != "v1" or len(elems[2]) == 0:
            return None
        return elems[2]
    
    def _read_to_buffers(self, file: TextIO, target: float, category: str, type: str, buffers: Dict[str, List[_SignalElem]]) -> float:
        last_time = 0
        while True:
            elems = file.readline().replace("\n", "").replace("\r", "").split(",")
            if len(elems) != 3:
                break
            signal_name = elems[0]
            time = float(elems[1])
            val = float(elems[2])
            last_time = max(last_time, time)
            if len(signal_name) != 0:
                signal_id = category + ":" + type + ":" + signal_name
                if signal_id not in buffers:
                    buffers[signal_id] = []
                buffers[signal_id].append(_SignalElem(time, val))
            if time >= target:
                break
        return last_time
    
    def _clip_buffers(self, target: float, buffers: Dict[str, List[_SignalElem]]):
        for id in buffers:
            buffers[id] = list(filter(lambda elem: elem.time >= target, buffers[id]))


class _SignalWriter:
    def __init__(self, out_signal_dir: str, script_id: str):
        type_dir = os.path.join(out_signal_dir, "pluginpython.dll")
        os.makedirs(type_dir)
        self._file: TextIO = open(os.path.join(type_dir, script_id + ".csv"), "w")
        self._file.write("Signal Table,v1," + script_id + "\n")

    def write(self, name: str, time: float, val: float):
        self._file.write(name + "," + str(time) + "," + str(val) + "\n")

    def close(self):
        self._file.close()


class _SampleLoader:
    def __init__(self, sample_csv_path: str, protocol: str):
        self._valid: bool = False
        self._file: TextIO | None = None
        self._protocol: str = ""
        self._offset: int = 0
        self._pair: List[bi.GeneralSample | None] = [None, None]
        file = open(sample_csv_path)
        offset = self._read_title(file)
        if offset is not None:
            self._valid = True
            self._file = file
            self._protocol = protocol
            self._offset = offset
        else:
            file.close()

    def load(self, target: float, max_gap: float) -> bi.GeneralSamplePair | None:
        if not self._valid or self._file is None:
            return None
        self._update_pair(self._file, target, self._protocol, self._offset, self._pair)
        pair = self._pair
        if pair[0] is None or pair[1] is None:
            return None
        dt1 = target - pair[0].time
        dt2 = pair[1].time - target
        sample_pair = bi.agency.create_general_sample_pair()
        if dt1 > max_gap:
            if dt2 <= max_gap:
                sample_pair.sample1 = sample_pair.sample2 = pair[1]
                sample_pair.weight1 = 1.0
            else:
                return None
        else:
            if dt2 <= max_gap:
                sample_pair.sample1 = pair[0]
                sample_pair.sample2 = pair[1]
                sample_pair.weight1 = dt2 / (dt1 + dt2)
                sample_pair.weight2 = dt1 / (dt1 + dt2)
            else:
                sample_pair.sample1 = sample_pair.sample2 = pair[0]
                sample_pair.weight1 = 1.0
        return sample_pair

    def close(self):
        if self._file is not None:
            self._file.close()
            self._file = None
        self._valid = False

    def _update_pair(self, file: TextIO, target: float, protocol: str, offset: int, pair: List[bi.GeneralSample | None]) -> None:
        while True:
            if pair[1] is not None and pair[1].time <= target:
                pair[0] = pair[1]
                pair[1] = None
            if pair[1] is None:
                pair[1] = self._read_row(file, offset, protocol)
            if pair[1] is None or pair[1].time > target:
                break

    def _read_row(self, file: TextIO, offset: int, protocol: str) -> bi.GeneralSample | None:
        elems = file.readline().replace("\n", "").replace("\r", "").split(",")
        if len(elems) <= offset:
            return None
        sample = bi.agency.create_general_sample()
        sample.protocol = protocol
        sample.time = float(elems[offset])
        sample.values = []
        for i in range(offset + 1, len(elems)):
            elem = elems[i]
            if elem == "na":
                sample.values.append(None)
            elif elem.isdecimal():
                sample.values.append(float(elem))
            else:
                sample.values.append(str(elem))
        sample.significant = len(sample.values)
        return sample

    def _read_title(self, file: TextIO) -> int | None:
        elems = file.readline().rstrip("\n").split(",")
        if "Time" not in elems:
            elems = file.readline().rstrip("\n").split(",")
            if "Time" not in elems:
                return None
        return elems.index("Time")


class _SampleWriter:
    def __init__(self, sample_csv_path: str, session: datetime.datetime, title: str):
        self._file: TextIO = open(sample_csv_path, "w")
        self._file.write("Session,Time," + title + "\n")
        self._session: datetime.datetime = session

    def write(self, sample: bi.GeneralSample):
        elems = list()
        elems.append(self._session.strftime("%Y%m%d-%H-%M-%S"))
        elems.append("%.6f" % sample.time)
        count = min(len(sample.values), sample.significant)
        for i in range(0, count):
            val = sample.values[i]
            if val is None:
                elems.append("na")
            else:
                elems.append(str(val))
        self._file.write(",".join(elems) + "\n")

    def close(self):
        self._file.close()


class _ReportWriter:
    def __init__(self, out_gen_dir: str, config: _ReportOutConfig):
        self._data: Any = None
        self._global_id: int = self._calc_global_id(config)
        graph_dir = os.path.join(out_gen_dir, "graph")
        if not os.path.exists(graph_dir):
            os.makedirs(graph_dir)
        self._path: str = os.path.join(graph_dir, config.title + ".csv")
        if config.type == _GraphType.SINGLE_VALUE:
            self._data = _SingleValueReportData(config)
        elif config.type == _GraphType.SCATTER_POINTS:
            self._data = _ScatterPointsReportData(config)
        elif config.type == _GraphType.HIST_AND_LINE:
            self._data = _HistAndLineReportData(config)
        elif config.type == _GraphType.MATRIX_TABLE:
            self._data = _MatrixTableReportData(config)
        elif config.type == _GraphType.LABEL_TABLE:
            self._data = _LabelTableReportData(config)
        

    def get_global_id(self) -> int:
        return self._global_id

    def update_report(self, parameters: Dict[str, Any]):
        if self._data is not None and hasattr(self._data, "update_report"):
            self._data.update_report(parameters)

    def save(self):

        if self._data is None:
            return
        
        file = open(self._path, "w")

        params_text = ""
        if self._data.params is not None and len(self._data.params) > 0:
            params_text = "," + ','.join(self._data.params)
        file.write("ASEva Report v2," + str(self._global_id) + "," + self._data.config.title + params_text + "\n")
        
        type_text = _GraphType.to_camel_string(self._data.config.type)
        configs_text = ""
        if self._data.config.configs is not None and len(self._data.config.configs) > 0:
            configs_text = "," + ','.join(self._data.config.configs)
        file.write(type_text + configs_text + "\n")

        file.write("NoValidation\n")

        file.write(','.join(self._data.config.column_titles) + "\n")

        for row in self._data.data:
            elems = []
            for col in row:
                elems.append(str(col))
            file.write(','.join(elems) + "\n")

        file.close()

    def _calc_global_id(self, config: _ReportOutConfig) -> int:

        if len(config.title) == 0 or len(config.column_titles) == 0:
            return 0
        
        target = _GraphType.to_camel_string(config.type)
        for c in config.configs:
            target = target + ":" + c
        target = target + ":" + config.title
        for t in config.column_titles:
            target = target + ":" + t
        bytes = bytearray(target, encoding='utf-8')

        hash = 0
        for b in bytes:
            hash = hash + b
            hash = hash & 0xffffffff
            hash = hash * 3
            hash = hash & 0xffffffff
        return hash & 0x7fffffff


class _StandaloneAgency:
    def __init__(self, out_gen_dir: str, global_params: Dict[str, Any], global_vars: Dict[str, Any]):
        self._out_gen_dir: str = out_gen_dir
        self._global_params: Dict[str, Any] = global_params
        self._global_vars: Dict[str, Any] = global_vars
        self._report_writers: Dict[int, _ReportWriter] = {}
        self._scene_titles: Dict[str, str] = {}
        self._scene_files: Dict[str, TextIO] = {}
        self._bustx_file: TextIO | None = None
        self._time: float = 0.0

    def close(self):
        for id in self._report_writers:
            self._report_writers[id].save()
        self._report_writers.clear()
        for id in self._scene_files:
            self._scene_files[id].close()
        self._scene_files.clear()
        if self._bustx_file is not None:
            self._bustx_file.close()
            self._bustx_file = None

    def init_with_config(self, out_gen_dir: str, report_out: List[_ReportOutConfig], scene_out: List[_SceneOutConfig]):
        for config in report_out:
            writer = _ReportWriter(out_gen_dir, config)
            global_id = writer.get_global_id()
            self._report_writers[global_id] = writer
            bi.report_id_table[config.id] = global_id
        for config in scene_out:
            self._scene_titles[config.id] = config.title

    def update_time(self, time: float):
        self._time = time

    def create_general_sample(self) -> bi.GeneralSample:
        return bi.GeneralSample()

    def create_general_sample_pair(self) -> bi.GeneralSamplePair:
        return bi.GeneralSamplePair()

    def create_bus_message(self) -> bi.BusMessage:
        return bi.BusMessage()

    def create_scene(self) -> bi.Scene:
        return bi.Scene()

    def get_global_param(self, key: str) -> Any:
        return self._global_params[key]

    def set_global_param(self, key: str, value: Any):
        self._global_params[key] = value

    def add_scene(self, sess: datetime.datetime, scene: bi.Scene):
        if scene.id not in self._scene_files:
            scene_dir = os.path.join(self._out_gen_dir, "scene")
            if not os.path.exists(scene_dir):
                os.makedirs(scene_dir)
            self._scene_files[scene.id] = open(os.path.join(scene_dir, scene.id + ".csv"), "w")
            self._scene_files[scene.id].write("Scene Table,v2\n")
            header = "Session,Start time,Time length,Scene type"
            if scene.id in self._scene_titles:
                header = header + "," + self._scene_titles[scene.id]
            self._scene_files[scene.id].write(header + "\n")
        elems = []
        elems.append(sess.strftime("%Y%m%d-%H-%M-%S"))
        elems.append(str(float(scene.begin)))
        elems.append(str(float(scene.length)))
        elems.append(scene.id)
        for prop in scene.properties:
            elems.append(str(prop))
        self._scene_files[scene.id].write(','.join(elems) + "\n")

    def update_report(self, sess: datetime.datetime, global_id: str, parameters: Dict[str, Any]):
        if global_id in self._report_writers:
            self._report_writers[global_id].update_report(parameters)

    def send_bus_message(self, msg: bi.BusMessage):
        if self._bustx_file is None:
            self._bustx_file = open(os.path.join(self._out_gen_dir, "send_bus_message.csv"), "w")
            self._bustx_file.write("Time,Channel,ID,Period,Data" + "\n")
        elems = []
        elems.append(str(self._time))
        elems.append(str(int(msg.channel)))
        elems.append(str(int(msg.id)))
        if msg.periodic:
            elems.append(str(int(msg.period)) + "ms")
        else:
            elems.append("once")
        for b in msg.data:
            elems.append(str(int(b)))
        self._bustx_file.write(','.join(elems) + "\n")

    def get_global_var(self, key: str) -> Any:
        return self._global_vars[key]

    def set_global_var(self, key: str, value: Any):
        self._global_vars[key] = value


class _StandaloneVars:
    init_ok: bool = False
    length: float = 0
    frequency: int = 0
    out_gen_dir: str = ""
    io_config: _ConfigIO | None = None
    signal_loaders: List[_SignalLoader] = []
    signal_writer: _SignalWriter | None = None
    sample_loaders: Dict[str, _SampleLoader] = {}
    sample_writers: Dict[str, _SampleWriter] = {}
    cur_loop_index: int = 0
    checkpoint: float = 0
    

def _find_in_gen_dir(output_dir: str, generation: str) -> str | None:
    for name in os.listdir(output_dir):
        gen_dir = os.path.join(output_dir, name)
        info_path = os.path.join(gen_dir, "info.xml")
        if not os.path.exists(info_path):
            continue
        gen_info = _GenerationInfo(info_path)
        if gen_info.gen_id == generation:
            return gen_dir
    return None


class _SampleInFileInfo:
    def __init__(self, path: str, protocol: str):
        self.path: str = path
        self.protocol: str = protocol


def _find_sample_in_file(in_sample_dir: str, protocol: str) -> _SampleInFileInfo | None:

    main = protocol
    ver = None
    channel = None

    at_index = main.rindex("@") if "@" in main else -1
    if at_index > 0:
        channel = main[at_index + 1:]
        main = main[0:at_index]
    
    barv_index = main.rindex("-v") if "-v" in main else -1
    if barv_index > 0:
        ver = int(main[barv_index + 2:])
        main = main[0:barv_index]

    if ver is None:
        test_path = os.path.join(in_sample_dir, protocol + ".csv")
        if os.path.exists(test_path):
            return _SampleInFileInfo(test_path, protocol)
    else:
        for v in range(1, ver + 1):
            target_protocol = main + "-v" + str(v)
            if channel is not None:
                target_protocol = target_protocol + "@" + channel
            test_path = os.path.join(in_sample_dir, target_protocol + ".csv")
            if os.path.exists(test_path):
                return _SampleInFileInfo(test_path, target_protocol)
    
    return None


def standalone_before_start(script_file: str, session_dir: str, generation: str, frequency: int, global_params: Dict[str, Any], global_vars: Dict[str, Any], globals: Dict[str, Any]) -> bool:
    """ 在运行Start脚本前的操作

    Args:
        script_file: str 脚本路径
        session_dir: str Session数据文件夹路径
        generation: str 输入Generation的ID
        frequency: int 脚本运行频率，单位Hz
        global_params: dict[str, str] 全局参数表
        global_vars: dict[str, str] 全局变量表
        globals: dict[str, Any] 脚本的变量对象
    Returns:
        bool 是否初始化成功
    """

    if frequency < 5 or frequency > 100:
        print("Invalid frequency.")
        return False

    meta_path = os.path.join(session_dir, "meta.xml")
    if not os.path.exists(meta_path):
        print("Session or meta.xml not found.")
        return False
    
    script_id = os.path.basename(script_file).split(".")[0]
    io_config_path = os.path.join(os.path.abspath("."), script_id + ".io.xml")
    if (not os.path.exists(io_config_path)):
        print(os.path.basename(io_config_path) + " not found.")
        return False

    output_dir = os.path.join(session_dir, "output")
    if not os.path.exists(output_dir):
        os.makedirs(output_dir)
    in_gen_dir = _find_in_gen_dir(output_dir, generation)
    
    out_gen_dir = os.path.join(output_dir, "python-" + datetime.datetime.now().strftime("%Y%m%d-%H-%M-%S"))
    os.makedirs(out_gen_dir)

    session_meta = _SessionMeta(meta_path)
    io_config = _ConfigIO(io_config_path, script_id, globals)

    bi.agency = _StandaloneAgency(out_gen_dir, global_params, global_vars)
    bi.global_public_path = os.path.join(os.path.dirname(session_dir), "public")
    bi.session_public_path = os.path.join(session_dir, "public")
    bi.interval = 1.0 / frequency
    bi.session = session_meta.session_id
    bi.template_postfix = ""

    signal_loaders = []
    if in_gen_dir is not None:
        in_signal_dir = os.path.join(in_gen_dir, "signal")
        for category in os.listdir(in_signal_dir):
            category_dir = os.path.join(in_signal_dir, category)
            for signal_csv_path in glob.glob(category_dir + "/*.csv"):
                signal_loaders.append(_SignalLoader(signal_csv_path, category))

    signal_writer = None
    if len(io_config.signal_out) > 0:
        out_signal_dir = os.path.join(out_gen_dir, "signal")
        signal_writer = _SignalWriter(out_signal_dir, script_id)

    sample_loaders = {}
    if in_gen_dir is not None:
        in_sample_dir = os.path.join(in_gen_dir, "sample")
        for protocol in io_config.sample_in:
            sample_in_info = _find_sample_in_file(in_sample_dir, protocol)
            if sample_in_info is not None:
                sample_loaders[sample_in_info.protocol] = _SampleLoader(sample_in_info.path, sample_in_info.protocol)

    sample_writers = {}
    out_sample_dir = os.path.join(out_gen_dir, "sample")
    if not os.path.exists(out_sample_dir):
        os.makedirs(out_sample_dir)
    for config in io_config.sample_out:
        sample_csv_path = os.path.join(out_sample_dir, config.protocol + ".csv")
        sample_writers[config.protocol] = _SampleWriter(sample_csv_path, bi.session, config.title)

    bi.agency.init_with_config(out_gen_dir, io_config.report_out, io_config.scene_out)

    _StandaloneVars.init_ok = True
    _StandaloneVars.length = session_meta.length
    _StandaloneVars.frequency = frequency
    _StandaloneVars.out_gen_dir = out_gen_dir
    _StandaloneVars.io_config = io_config
    _StandaloneVars.signal_loaders = signal_loaders
    _StandaloneVars.signal_writer = signal_writer
    _StandaloneVars.sample_loaders = sample_loaders
    _StandaloneVars.sample_writers = sample_writers
    _StandaloneVars.cur_loop_index = 1
    _StandaloneVars.checkpoint = 5

    return True


def standalone_before_loop(globals: Dict[str, Any]):
    """ 在运行Loop脚本前的操作

    Args:
        globals: dict[str, Any] 脚本的变量对象
    """

    if not _StandaloneVars.init_ok or _StandaloneVars.io_config is None:
        return

    bi.time = float(_StandaloneVars.cur_loop_index) / _StandaloneVars.frequency

    bi.agency.update_time(bi.time)

    for loader in _StandaloneVars.signal_loaders:
        loader.load(bi.time, 1)

    for config in _StandaloneVars.io_config.signal_in:
        found = False
        for loader in _StandaloneVars.signal_loaders:
            if config.source in loader.signal_ids():
                globals[config.name] = loader.get_val(bi.time, config.source, config.nearest, config.default_val)
                found = True
        if not found:
            globals[config.name] = config.default_val

    bi.input_samples.clear()
    for protocol in _StandaloneVars.sample_loaders:
        sample_pair = _StandaloneVars.sample_loaders[protocol].load(bi.time, 1)
        if sample_pair is not None:
            bi.input_samples[protocol] = sample_pair

    bi.output_samples.clear()


def standalone_after_loop(globals: Dict[str, Any]) -> bool:
    """ 在运行Loop脚本后的操作

    Args:
        globals: dict[str, Any] 脚本的变量对象
    Returns:
        bool 未结束(False表示循环结束)
    """

    if not _StandaloneVars.init_ok or _StandaloneVars.io_config is None:
        return False

    if _StandaloneVars.signal_writer is not None:
        for name in _StandaloneVars.io_config.signal_out:
            if name in globals:
                val = float(globals[name])
                _StandaloneVars.signal_writer.write(name, bi.time, val)
            pass

    for protocol in bi.output_samples:
        if protocol not in _StandaloneVars.sample_writers:
            continue
        samples = bi.output_samples[protocol]
        if not isinstance(samples, list):
            samples = [samples]
        for sample in samples:
            _StandaloneVars.sample_writers[protocol].write(sample)
    
    if bi.time >= _StandaloneVars.checkpoint:
        print("Progress: " + str(round(bi.time, 1)) + "s, " + str(round(100.0 * _StandaloneVars.checkpoint / _StandaloneVars.length, 2)) + "%")
        _StandaloneVars.checkpoint = _StandaloneVars.checkpoint + 5
    
    _StandaloneVars.cur_loop_index = _StandaloneVars.cur_loop_index + 1
    return float(_StandaloneVars.cur_loop_index) / _StandaloneVars.frequency <= _StandaloneVars.length


def standalone_after_end():
    """ 在运行End脚本后的操作 """

    if not _StandaloneVars.init_ok:
        return

    bi.agency.close()

    for loader in _StandaloneVars.signal_loaders:
        loader.close()
    _StandaloneVars.signal_loaders.clear()

    if _StandaloneVars.signal_writer is not None:
        _StandaloneVars.signal_writer.close()
    _StandaloneVars.signal_writer = None

    for protocol in _StandaloneVars.sample_loaders:
        _StandaloneVars.sample_loaders[protocol].close()
    _StandaloneVars.sample_loaders.clear()

    for protocol in _StandaloneVars.sample_writers:
        _StandaloneVars.sample_writers[protocol].close()
    _StandaloneVars.sample_writers.clear()

    _StandaloneVars.io_config = None
    _StandaloneVars.init_ok = False
    
    